package com.concurrent.frame.pojo;

import com.concurrent.frame.core.ResourceRegisterPool;
import com.concurrent.frame.mark.ThreadSafe;
import com.concurrent.frame.processor.task.AbsTaskProConProcessor;
import com.concurrent.frame.processor.task.ITaskProcessor;
import com.concurrent.frame.processor.complete.IJobComplete;
import com.concurrent.frame.processor.complete.ITaskComplete;
import com.concurrent.frame.util.Assert;
import com.sun.javafx.iio.ImageFormatDescription;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 工作实体类
 * 目的: 提交给框架执行
 */
@ThreadSafe
public class JobInfo<R> {
    //区分唯一工作
    private final String jobName;
    //工作的任务个数
    private final int jobLength;
    //工作任务的处理器
    private  ITaskProcessor<?,?> taskProcessor;
    //生产者消费者模式
    private CompletionService<TaskResult> completeService;
     //任务完成的处理器
    private  ITaskComplete<R> taskComplete;
    //工作完成处理器
    private  IJobComplete<R> jobComplete;

    //工作完成的处理器
    //成功任务个数
    private  AtomicInteger successCount;
    //任务总数
    private  AtomicInteger taskProcessorCount;
    //阻塞队列: 任务处理结果
    //双端队列: 头部获取结果,尾部放入结果,减少线程冲突
    private  LinkedBlockingDeque<TaskResult<R>> taskDetailQueue ;
    //是否开启缓存，保存结果
    private final boolean isCache;
    //过期时间
    private final Long expireTime;

    public String getJobName() {
        return jobName;
    }

    public JobInfo(String jobName, int jobLength, boolean isCache, Long expireTime) {
        this.jobName = jobName;
        this.jobLength = jobLength;
        this.isCache = isCache;
        this.successCount=new AtomicInteger(0);
        this.taskProcessorCount=new AtomicInteger(0);
        if (isCache)this.taskDetailQueue = new LinkedBlockingDeque<TaskResult<R>>(jobLength);
        this.expireTime = expireTime;
    }

    public CompletionService<TaskResult> getCompleteService() {
        return completeService;
    }

    public Integer getSuccessAccount() {
        return successCount.get();
    }
    public Integer getFailAccount() {
        return taskProcessorCount.get()-successCount.get();
    }

    /**
     * 获取任务进度
     * Success:成功个数
     * Current:进行个数
     * JobLength:任务总长度
     * @return
     */
    public String getTotalProcess(){
        return "Success["+successCount.get()+"]/Current["+taskProcessorCount.get()+"]/JobLength["+jobLength+"]";
    }
    public Integer getTaskProcessorAccount() {
        return taskProcessorCount.get();
    }

    /**
     * //线程安全的方式，发布阻塞队列
     * @return
     */
    public List<TaskResult<R>> getTaskDetail(){
        List<TaskResult<R>> taskList=new LinkedList<TaskResult<R>>();
        try {
            Assert.isNotNULL(taskDetailQueue,"阻塞队列没有初始化！");
            TaskResult<R> taskResult;
            while((taskResult=taskDetailQueue.pollFirst())!=null){
                taskList.add(taskResult);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return  taskList;
    }

    /**
     * 添加结果到阻塞队列
     * 并发安全问题:
     *          不用加锁,保证最终一致性,就能满足业务需求
     * @param result
     */
    public void addTaskResult(TaskResult<R> result){
        if (TaskResultType.Success.equals(result.getResultType() )){
            successCount.incrementAndGet();
        }
        taskProcessorCount.incrementAndGet();
        if (isCache)taskDetailQueue.addLast(result);//添加结果到阻塞队列
        //任务完成的 处理器
        if (taskComplete!=null){
            taskComplete.taskComplete(this,result);
        }

        if (taskProcessorCount.get()==jobLength){//任务已经处理完成!
            if (isCache&&expireTime>0){
                ResourceRegisterPool.getCheckJobProcessor().putJob(jobName,expireTime);//延迟队列,保存一定时间,定时销毁.
            }else{
                ResourceRegisterPool.getMap().remove(jobName);
            }
            //工作完成的处理器
            if (jobComplete!=null){
                jobComplete.taskComplete(this);
            }
        }
    }

    public void setTaskComplete(ITaskComplete<R> taskComplete) {
        if (this.taskComplete!=null){
            try {
                throw new Exception("JobInfo.TaskComplete已经存在，请勿重复设置！");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            this.taskComplete= taskComplete;
        }
    }

    public void setJobComplete(IJobComplete<R> jobComplete) {
        if (this.jobComplete!=null){
            try {
                throw new Exception("JobInfo.jobComplete已经存在，请勿重复设置！");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            this.jobComplete= jobComplete;
        }

    }

    public void setTaskProcessor(ITaskProcessor<?, ?> taskProcessor) {
        if (this.taskProcessor!=null){
            try {
                throw new Exception("JobInfo.taskProcessor已经存在，请勿重复设置！");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            this.taskProcessor = taskProcessor;
            if (taskProcessor instanceof AbsTaskProConProcessor){
                this.completeService = new ExecutorCompletionService<TaskResult>(ResourceRegisterPool.getTaskExecutor());
            }
        }
    }

    public ITaskProcessor<?, ?> getTaskProcessor() {
        return taskProcessor;
    }
}
